package com.wolf.etl.core.dialect;

import com.wolf.etl.constant.SqlConstant;
import com.wolf.etl.core.ETLManager;
import com.wolf.etl.core.filter.AbstractFilter;
import com.wolf.etl.enumeration.DIALECT;
import com.wolf.etl.enumeration.SQL_FORMULA;
import com.wolf.etl.model.*;
import com.wolf.etl.vo.BuildSqlParam;
import com.wolf.etl.vo.ColumnDesc;
import com.wolf.exception.WolfException;
import com.wolf.util.DateUtil;
import com.wolf.vo.BaseResult;
import com.wolf.vo.KeyValuePair;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

/**
 * @author sdyang
 * @create 2019-06-13 10:37
 **/
public abstract class AbstractDialect implements SqlConstant {

    private static Logger logger = LoggerFactory.getLogger(AbstractDialect.class);

    private static final String logger_prefix = "wolf-etl#com.wolf.etl.core.dialect#AbstractDialect";

    public static final String TRANSFORM_CLASSNAME_FORMAT = "%s.%s";

    protected static DIALECT dialect;

    protected AbstractFilter abstractFilter;

    protected Long datasourceId;

    protected List<KeyValuePair> sqls = new ArrayList<KeyValuePair>();

    protected ETLManager manager;

    public ETLManager getManager() {
        return manager;
    }

    public void setManager(ETLManager manager) {
        this.manager = manager;
    }


    protected List<Set<String>> columnTypes = new ArrayList<>();

    public List<Set<String>> getColumnTypes() {
        if (CollectionUtils.isEmpty(columnTypes)) {
            List<EtlTransformFieldTypeModel> fieldTypes = manager.findByDialect(dialect);
            for (EtlTransformFieldTypeModel t : fieldTypes) {
                String[] types = t.getField_type().split("#");
                columnTypes.add(new HashSet<>(Arrays.asList(types)));
            }
        }
        return columnTypes;
    }

    public boolean checkColumnType(String srcType, String targetType) {
        for (Set<String> types : columnTypes) {
            if (types.contains(srcType.toLowerCase())) {
                if (types.contains(targetType.toLowerCase())) {
                    return true;
                } else {
                    return false;
                }
            }
        }
        return false;
    }

//    protected Map<String, Set<String>> columnTypeMap = new HashMap<String, Set<String>>();
//
//    public Map<String, Set<String>> initColumnType() {
//        List<EtlTransformFieldTypeModel> vos = manager.findByDialect(dialect);
//        if (CollectionUtils.isEmpty(vos)) {
//            return columnTypeMap;
//        }
//        for (EtlTransformFieldTypeModel vo : vos) {
//            if (columnTypeMap.get(vo.getSource_field_type()) == null) {
//                Set<String> targets = new HashSet<>();
//                targets.add(vo.getTarget_field_type().toLowerCase());
//                columnTypeMap.put(vo.getSource_field_type(), targets);
//            } else {
//                columnTypeMap.get(vo.getSource_field_type()).add(vo.getTarget_field_type().toLowerCase());
//            }
//        }
//        return columnTypeMap;
//    }

//    public Map<String, Set<String>> getColumnTypeMap() {
//        if (columnTypeMap == null || columnTypeMap.size() == 0) {
//            columnTypeMap = initColumnType();
//        }
//        return columnTypeMap;
//    }

    public String getCustomFunction(String schema){
        return null;
    }

    public static DIALECT getDialect() {
        return dialect;
    }

    public static void setDialect(DIALECT dialect) {
        AbstractDialect.dialect = dialect;
    }

    public static AbstractDialect getInstance(ETLManager manager, DIALECT dialect) {

        AbstractDialect abstractDialect = null;
        String packageName = getPackageName();
        Class<?> clazz = null;
        try {
            clazz = Class.forName(String.format(TRANSFORM_CLASSNAME_FORMAT, packageName, dialect.getClassName()));
            abstractDialect = (AbstractDialect) manager.getBeanManager().createBean(clazz);
            abstractDialect.manager = manager;
            setDialect(dialect);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
        return abstractDialect;

    }


    public void addSql(KeyValuePair keyValuePair) {
        sqls.add(keyValuePair);
    }

    protected String getValue(Object object) {
        if (object == null) {
            return "";
        }
        return object.toString();
    }

    protected String buildOrderBy(String alias, EtlTableOrderbyModel orderby) {
        if (orderby == null) {
            return "";
        }

        StringBuffer orderbySql = new StringBuffer();
        orderbySql.append(ORDER_BY);

        orderbySql.append(String.format(FORMAT_PREFFIX_FIELD, alias, orderby.getColumn_name()));
        if (orderby.getIs_asc()) {
            orderbySql.append(ORDER_BY_ASC);
        } else {
            orderbySql.append(ORDER_BY_DESC);
        }

        return orderbySql.toString();
    }

    public abstract String buildPaginationSql(String originalSql, int offset, int limit);

    public String buildCount(BuildSqlParam param){

        // where语句
        StringBuffer whereSql = new StringBuffer(WHERE);
        for (EtlTableFilterModel filter : param.getFilters()) {
            whereSql.append(buildWhereSql(param.getAliasMap().get(filter.getEtl_table_id()), filter));
        }

        // orderby语句
        StringBuffer orderbySql = new StringBuffer();
        for (EtlTableOrderbyModel orderby : param.getOrderbys()) {
            orderbySql.append(buildOrderBy(param.getAliasMap().get(orderby.getEtl_table_id()), orderby));
        }

        String result = String.format(FORMAT_COUNT_SQL,  getTableWithAlias(getTableName(param.getSrcTable()), param.getAliasMap().get(param.getSrcTable().getId())), whereSql);
        logger.info("查询总数sql："+result);
        return result;
    }

    public String buildSelect(BuildSqlParam param) {
        StringBuffer sql = new StringBuffer();

        for (EtlTransformDefineModel transform : param.getTransformDefineModelList()) {
            sql.append(getFieldWithPrefix(param.getAliasMap().get(transform.getSrc_table_id()), transform.getSrc_column_name())).append(COMMA);
        }

        String columnSql =sql.substring(0, sql.length() - 1);

        // where语句
        StringBuffer whereSql = new StringBuffer(WHERE);
        for (EtlTableFilterModel filter : param.getFilters()) {
            whereSql.append(buildWhereSql(param.getAliasMap().get(filter.getEtl_table_id()), filter));
        }

        // orderby语句
        StringBuffer orderbySql = new StringBuffer();
        for (EtlTableOrderbyModel orderby : param.getOrderbys()) {
            orderbySql.append(buildOrderBy(param.getAliasMap().get(orderby.getEtl_table_id()), orderby));
        }

        String result = String.format(FORMAT_SELECT_SQL, columnSql, getTableWithAlias(getTableName(param.getSrcTable()), param.getAliasMap().get(param.getSrcTable().getId())), whereSql, orderbySql);

        // 分页语句
        result = buildPaginationSql(result, param.getOffset(), param.getFetch_max_count());

        System.out.println(result);
        return result;
    }

    public String buildInsert(EtlTableModel table, Map<String, ColumnDesc> fieldData) {
        if (fieldData == null || fieldData.size() == 0) {
            return "";
        }

        String fieldSql = "";
        String valueSql = "";
        StringBuffer fields = new StringBuffer();
        StringBuffer values = new StringBuffer();

        for (Map.Entry<String, ColumnDesc> entry : fieldData.entrySet()) {
            Object obj = entry.getValue().getObj();
            if (obj == null) {
                continue;
            }
            fields.append(entry.getKey()).append(COMMA);
            if (entry.getValue().isNumberType()) {
                values.append(obj.toString()).append(COMMA);
            } else if (entry.getValue().isDateType()) {
                Date date = (Date) obj;
                values.append(SINGLE_QUOTE).append(DateUtil.toString(date)).append(SINGLE_QUOTE).append(COMMA);
            } else if (entry.getValue().isFormula()) {
                values.append(obj.toString()).append(COMMA);
            } else
                values.append(SINGLE_QUOTE).append(obj).append(SINGLE_QUOTE).append(COMMA);
        }


        if (StringUtils.isNotEmpty(fields) && fields.toString().endsWith(COMMA)) {
            fieldSql = fields.substring(0, fields.length() - 1);
        }
        if (StringUtils.isNotEmpty(values) && values.toString().endsWith(COMMA)) {
            valueSql = values.substring(0, values.length() - 1);
        }

        String result = String.format(FORMAT_INSERT_SQL, getTableName(table), fieldSql, valueSql);
        return result;
    }

    public abstract String buildInsertSelect(EtlTableModel table, String targerTableName, Map<String, String> fields);

    public String buildOrderPart() {
        return "";
    }

    public String getTableName(EtlTableModel table) {
        return table.getName();
    }

    private String buildFieldParts(String alias, KeyValuePair... kvps) {
        StringBuffer sb = new StringBuffer();
        int index = 0;
        for (KeyValuePair kvp : kvps) {
            String str = String.format(FORMAT_UPDATE_FIELD,
                    String.format(FORMAT_PREFFIX_FIELD, alias, kvp.getKey()),
                    getValue(kvp.getValue()));
            sb.append(str);
            if (index < kvps.length - 1) {
                sb.append(COMMA);
            }
            index++;
        }
        return sb.toString();
    }

    private String getValue(String data) {
        if (StringUtils.isNumeric(data)) {
            return data;
        }
        return String.format(FORMAT_STRING, data);
    }

    protected String buildWhereSql(String alias, EtlTableFilterModel filter) {
        if (filter == null) {
            return "";
        }
        StringBuffer sql = new StringBuffer();

        String field = String.format(FORMAT_PREFFIX_FIELD, alias, filter.getColumn_name());

        SQL_FORMULA sqlFormula = SQL_FORMULA.get(filter.getFormula());
        AbstractFilter abstractFilter = AbstractFilter.getInstance(manager, sqlFormula);
        abstractFilter.setAbstractDialect(this);
        BaseResult result = abstractFilter.filter(field, filter.getColumn_value1(), filter.getColumn_value2());
        if (result.isSucc()) {
            sql.append(result.getMessage());
        } else {
            throw new WolfException(String.format("Can't build where sql: %s", filter.getFormula()));
        }

        return sql.toString();
    }

    private String findSqlByTabId(Long tabId) {
        for (KeyValuePair kvp : sqls) {
            if (tabId == Long.parseLong(kvp.getKey())) {
                return kvp.getValue();
            }
        }
        throw new WolfException(String.format("Can't find any sql info by the table id[%s]", tabId));
    }

    protected String getTableAlias(EtlTableModel model) {
        if (StringUtils.isEmpty(model.getName())) {
            throw new WolfException(String.format("ETL table info[id=>%] it's table name is empty.", model.getId()));
        }

        String alias = model.getName();
        if (!StringUtils.isEmpty(model.getAlias())) {
            alias = model.getAlias();
        }
        return alias;
    }

    protected String getFieldWithPrefix(String alias, String field) {
        return String.format(FORMAT_PREFFIX_FIELD, alias, field);
    }

    private String buildWhereExistsParts(EtlTableModel table) {
        String whereStr = " where exists(%s)";
        String sql = findSqlByTabId(table.getId());
        sql = String.format(sql, " 1 ");

        String alias = getTableAlias(table);
        sql = sql + String.format(FORMAT_AND_FIELD,
                String.format(FORMAT_PREFFIX_FIELD, alias, table.getPrimary_key()),
                String.format(FORMAT_PREFFIX_FIELD, "t", table.getPrimary_key()));
        sql = String.format(whereStr, sql);

        return sql;
    }

    protected String appendAnd(String sql, String sqlStr) {
        String OPEN_PARENTHESIS = "(";
        String CLOSE_PARENTHESIS = ")";
        String AND_STR = "and";
        String AND_STR_1 = " 1=1 and ";
        String WHERE_STR = "where";

        int index = sql.lastIndexOf(WHERE_STR);

        //no exist word of 'where'
        if (index == -1) {
            sqlStr = WHERE_STR + " " + sqlStr;
            return sqlStr;
        }

        //exist word of 'where'
        //don't exist '(' and ')'
        //don't exist 'and'
        String leftStr = sql.substring(0, index);
        String rightStr = sql.substring(index + 1, sql.length());
        if (leftStr.indexOf(OPEN_PARENTHESIS) == -1 &&
                rightStr.indexOf(CLOSE_PARENTHESIS) == -1) {
            if (rightStr.indexOf(AND_STR) == -1) {
                sqlStr = AND_STR_1 + sqlStr;
                return sqlStr;
            }
            return sqlStr;
        }

        int index1 = leftStr.indexOf(OPEN_PARENTHESIS);
        int index2 = rightStr.lastIndexOf(CLOSE_PARENTHESIS);

        //exist word of 'where'
        //do exist '(' and ')'
        //position index of ')' > '(' and 'where' > ')'
        //don't exist 'and'
        if (index1 != -1
                && index2 != -1
                && index2 > index1
                && index > index2) {
            if (rightStr.indexOf(AND_STR) == -1) {
                sqlStr = AND_STR_1 + sqlStr;
                return sqlStr;
            }
            return sqlStr;
        }
        return "";
    }

    public String buildUpdate(EtlTableModel table, KeyValuePair[] keyValuePairs, List<EtlTableFilterModel> filters) {
        StringBuffer whereSql = new StringBuffer();
        for (EtlTableFilterModel filter : filters) {
            whereSql.append(buildWhereSql(table.getAlias(), filter));
        }
        String sql = String.format(FORMAT_UPDATE_SQL,
                table.getName(),
                table.getAlias(),
                buildFieldParts(table.getAlias(), keyValuePairs),
                whereSql);

        return sql;
    }

    protected String getTableWithAlias(String table, String alias) {
        return String.format(FORMAT_TABLE_ALIAS, table, alias);
    }

    private static final String getPackageName() {
        return new Object() {
            public String getClassForStatic() {
                return this.getClass().getPackage().getName();
            }
        }.getClassForStatic();
    }

    public Long getDatasourceId() {
        return datasourceId;
    }

    public void setDatasourceId(Long datasourceId) {
        this.datasourceId = datasourceId;
    }
}
